package chapter4.eg5;

import java.util.Collection;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {

  private final KafkaConsumer<String, String> consumer;

  public SaveOffsetsOnRebalance(KafkaConsumer<String, String> consumer) {
    this.consumer = consumer;
  }

  @Override
  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    commitDBTransaction();
  }

  private void commitDBTransaction() {}

  @Override
  public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    for (TopicPartition partition : partitions) {
      consumer.seek(partition, getOffsetFromDB(partition));
    }
  }

  private long getOffsetFromDB(TopicPartition partition) {
    return -1;
  }
}
